package com.sunwayworld.basemodule.common.eptstorage.acqconfig.net;

import com.sunwayworld.basemodule.common.eptstorage.acqconfig.bean.LimsEquiptAcqConfigInstBean;
import com.sunwayworld.basemodule.common.eptstorage.util.Util;
import com.sunwayworld.cloud.module.equipt.acqconfig.bean.CoreEquiptAcqConfigBean;
import com.sunwayworld.framework.constant.Constant;
import com.sunwayworld.framework.exception.InvalidDataException;
import com.sunwayworld.framework.utils.StringUtils;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * @auther zhangs
 * @description
 * @date 2022/3/4
 */
public class NetEptTcpSender extends NetEptSender{
    public ChannelFuture channelFuture;
    private NioEventLoopGroup group = null;
    private Object msg = null;
    private CountDownLatch waitReadLatch = null;
    private CountDownLatch waitConnectLatch = null;
    public Object getMsg() {
        return msg;
    }
    public void setMsg(Object msg) {
        this.msg = msg;
    }

    public NetEptTcpSender(CoreEquiptAcqConfigBean eptAcqBean) {
        super(eptAcqBean);
    }

    @Override
    public boolean isConnect() {
        return connect;
    }

    @Override
    public boolean getConnection() {
        CoreEquiptAcqConfigBean eptAcqBean = getEptAcqBean();
        String host = eptAcqBean.getIp();
        int port = Integer.parseInt(eptAcqBean.getPort());
        boolean connect = isConnect();
        if(!connect){
            getLogger().info(Thread.currentThread().getName() + "---try getConnection for remoteAddress={}",(host+":"+port));
            //尝试连接一次
            tryConnect();
        }else{
            getLogger().info(Thread.currentThread().getName() + "---reuse getConnection for remoteAddress={}",(host+":"+port));
        }
        return isConnect();
    }

    private void tryConnect() {
        CoreEquiptAcqConfigBean eptAcqBean = getEptAcqBean();
        String host = eptAcqBean.getIp();
        int port = Integer.parseInt(eptAcqBean.getPort());
        group = new NioEventLoopGroup();
        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(group);
            bootstrap.channel(NioSocketChannel.class);
            //保持长连接状态
            bootstrap.option(ChannelOption.SO_KEEPALIVE, true);
            bootstrap.option(ChannelOption.TCP_NODELAY, true);
            bootstrap.handler(new ChannelInitializer() {
                @Override
                protected void initChannel(Channel channel) {
                    channel.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                        @Override
                        public void channelRead(ChannelHandlerContext ctx, Object msg) {
                            setMsg(msg);
                            if(waitReadLatch!=null){
                                waitReadLatch.countDown();
                            }
                        }

                        @Override
                        public void channelActive(ChannelHandlerContext ctx) {
                            setConnect(true);
                            getLogger().info(Thread.currentThread().getName() + "---connectSuccess--------->remoteAddress={}", ctx.channel().remoteAddress());
                            cacheSender();
                            waitConnectLatch.countDown();
                            startAsyncSequenceReadQueueAndSendWhenActive();
                        }

                        @Override
                        public void channelInactive(ChannelHandlerContext ctx) throws Exception{
                            getLogger().error(Thread.currentThread().getName() + "---connectFail--------->remoteAddress={}", channelFuture.channel().remoteAddress());
                            ctx.close();
                            close();
                        }
                    });
//                    channel.pipeline().addLast(new IdleStateHandler(0,0,5){
//                        @Override
//                        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
//                            super.userEventTriggered(ctx,evt);
//                            getLogger().error(Thread.currentThread().getName() + "---IdleStateHandler开始发送--------->remoteAddress={}", channelFuture.channel().remoteAddress());
//                        }
//                    });
                }
            });
            String threadName = Thread.currentThread().getName();
            getLogger().info(threadName+"---start connect--------->remoteAddress={}",(host+":"+port));
            waitConnectLatch = new CountDownLatch(1);
            channelFuture = bootstrap.connect(host,port);
//            channelFuture.sync().await(waitConnectSecond, TimeUnit.SECONDS); //设置超时2s连接成功
            boolean await = waitConnectLatch.await(waitConnectSecond,TimeUnit.SECONDS);//同步连接操作，超时时间2秒
            //将配置缓存到内存中
            if(!NetEptSenderHelper.cacheEptAcqConfigMap.containsKey(host+":"+port)){
                NetEptSenderHelper.cacheEptAcqConfigMap.put(host+":"+port,eptAcqBean);
            }
        }catch (Throwable s){
            getLogger().error(Thread.currentThread().getName()+"---connectFail--------->remoteAddress={},cause={}",host+":"+port,s);
        }finally {
//            close();
        }
    }

    @Override
    public  String send(LimsEquiptAcqConfigInstBean configInstBean) {
        CoreEquiptAcqConfigBean eptAcqBean = getEptAcqBean();
        String host = eptAcqBean.getIp();
        int port = Integer.parseInt(eptAcqBean.getPort());

        if(configInstBean==null || !isConnect()){
            getLogger().error(Thread.currentThread().getName()+"---sendFail--------->remoteAddress={},cause={}",host+":"+port,"未连接");
            close();
            throw new InvalidDataException("连接已断开，发送指令失败");
        }

        getLogger().info(Thread.currentThread().getName()+"---sendMsg={},to={}",configInstBean.getInstruction(),(host+":"+port));

        ByteBuf byteBuf;
        if (StringUtils.equalsIgnoreCase(Constant.YES, configInstBean.getHixFlag())) {
            byte[] bytes = Util.hixStringToByteArray(configInstBean.getInstruction());
            byteBuf = Util.convert2Buffer(bytes);
        } else {
            //字符串命令都要结束符 一般为\n
            String msg = configInstBean.getInstruction()+System.getProperty("line.separator");
            byteBuf = Util.convert2Buffer(msg);
        }
        sendMsg(byteBuf);

        //最多等待2s,如果正常读了，则返回true
        try {
            waitReadLatch = new CountDownLatch(1);
            long t1 = System.currentTimeMillis();
            boolean await = waitReadLatch.await(readWaitSecond, TimeUnit.SECONDS);
            long t2 = System.currentTimeMillis();
            getLogger().info("等待读取数据用时：{},{}",(t2-t1),(host+":"+port));
            String result = "";
            if(await){
                getLogger().info("正常读取数据成功：{},{}",msg,(host+":"+port));
                Object msg = getMsg();
                if(msg!=null){
                    result = insertMsg(configInstBean,msg);
                }
            }else{
                getLogger().info("读取数据超时：{}",(host+":"+port));
                waitReadLatch.countDown();
                //未及时返回消息，可能是网络故障，关闭连接
                close();
                throw new InvalidDataException("读取数据超时:"+host+":"+port);
            }
            return result;
        } catch (InterruptedException e) {
            e.printStackTrace();
            throw new InvalidDataException(e.getMessage());
        }
    }

    @Override
    public void close() {
        getLogger().error("-----close---------");
        getLogger().info("为设备={},,close",getKey(getEptAcqBean()));
        setConnect(false);
        if(channelFuture!=null){
            channelFuture.channel().close();
        }
        if(group!=null){
            group.shutdownGracefully();
        }
        NetEptSenderHelper.removeNetEptSender(getEptAcqBean());
    }

    private void sendMsg(ByteBuf byteBuf){
        if(isConnect()){
            channelFuture.channel().writeAndFlush(byteBuf);
        }else{
            close();
        }
    }
}
